package producer

import (
	"context"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/kafka-service/service/connect"
	"gitee.com/zaper/kafka-service/service/producer/producer_model"
	"gitee.com/zaper/kafka-service/util"
	"gitee.com/zaper/private-logger"
	"github.com/segmentio/kafka-go"
	"sync"
	"time"
)

var (
	smBind     sync.Mutex
	smProducer sync.Mutex
)

type ProducerAgent struct {
	// 生产者需要归属当前实例，否则一个关闭全部不可用
	writer *kafka.Writer
}

func NewProducerAgent() *ProducerAgent {
	return &ProducerAgent{}
}

func (producerAgent *ProducerAgent) ProducerBind(model *producer_model.ProducerModel) {
	smBind.Lock()
	defer smBind.Unlock()
	producerAgent.writer = &kafka.Writer{
		Addr:         kafka.TCP(connect.Brokers...),
		Topic:        model.Topic,
		BatchSize:    util.OrDefault(model.BatchSize, 1), // 关键配置：每条消息立即发送
		ReadTimeout:  util.OrDefault(model.ReadTimeout, 10) * time.Second,
		WriteTimeout: util.OrDefault(model.WriteTimeout, 10) * time.Second,
		Balancer:     &kafka.LeastBytes{},
		RequiredAcks: kafka.RequireAll,
	}
}

func (producerAgent *ProducerAgent) ProducerMess(mess *model.MessModel) error {
	ctx := context.Background()
	defer func() {
		err := recover()
		if err != nil {
			logger.Errorf("Producer producer message paict is %s", err)
		}
	}()
	message := kafka.Message{
		Key:   mess.Key,
		Value: mess.Value,
	}
	smProducer.Lock()
	err := producerAgent.writer.WriteMessages(ctx, message)
	defer smProducer.Unlock()
	if err != nil {
		logger.Errorf("Producer mess is fail, err is %s", err)
		return err
	}
	return nil
}

func (producerAgent *ProducerAgent) Close() error {
	if producerAgent.writer != nil {
		err := producerAgent.writer.Close()
		if err != nil {
			return err
		}
	}
	return nil
}
